package com.example.testkafkaproto.producer.controller;

import com.alibaba.fastjson.JSON;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

@RestController
@RequestMapping("/producer")
public class ProducerController {

    private static final Logger logger = LoggerFactory.getLogger(ProducerController.class);

    @Autowired
    private KafkaProducer kafkaProducer;

    /**
     * 发送后即忘
     * @param topic
     * @param value
     * @return
     */
    @RequestMapping("/sendOnly")
    @ResponseBody
    public Map<String,Object> sendOnly(String topic,String value) {

        Map<String,Object> appResult = new HashMap<>();
        ProducerRecord<String,String> producerRecord = new ProducerRecord(topic, UUID.randomUUID().toString(),value);
        kafkaProducer.send(producerRecord);
        appResult.put("status","SUCCESS");
        return appResult;
    }

    /**
     * 同步发送
     * @param topic
     * @param value
     * @return
     */
    @RequestMapping("/sendSync")
    @ResponseBody
    public Map<String,Object> sendSync(String topic,String value) {

        Map<String,Object> appResult = new HashMap<>();
        ProducerRecord<String,String> producerRecord = new ProducerRecord(topic, UUID.randomUUID().toString(),value);
        Future<RecordMetadata> future = kafkaProducer.send(producerRecord);
        try {
            RecordMetadata recordMetadata = future.get();
            System.out.println("发送消息完毕,RecordMetadata:topic=" + recordMetadata.topic() + ",partition=" + recordMetadata.partition() + ",offset=" + recordMetadata.offset() + ",timestamp=" + recordMetadata.timestamp());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        appResult.put("status","SUCCESS");
        return appResult;
    }

    /**
     * 异步发送
     * @param topic
     * @param value
     * @return
     */
    @RequestMapping("/sendAsync")
    @ResponseBody
    public Map<String,Object> sendAsync(String topic,String value) {

        Map<String,Object> appResult = new HashMap<>();
        ProducerRecord<String,String> producerRecord = new ProducerRecord(topic, UUID.randomUUID().toString(),value);
        kafkaProducer.send(producerRecord, new Callback() {
            @Override
            public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                System.out.println("发送消息完毕,RecordMetadata:topic=" + recordMetadata.topic() + ",partition=" + recordMetadata.partition() + ",offset=" + recordMetadata.offset() + ",timestamp=" + recordMetadata.timestamp());
            }
        });
        appResult.put("status","SUCCESS");
        return appResult;
    }

    /**
     * 消息无key,轮询分区发送
     * @param topic
     * @param value
     * @return
     */
    @RequestMapping("/sendWithoutKey")
    @ResponseBody
    public Map<String,Object> sendWithoutKey(String topic,String value) {

        Map<String,Object> appResult = new HashMap<>();
        ProducerRecord<String,String> producerRecord = new ProducerRecord(topic,value);
        kafkaProducer.send(producerRecord, new Callback() {
            @Override
            public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                System.out.println("发送消息完毕,RecordMetadata:topic=" + recordMetadata.topic() + ",partition=" + recordMetadata.partition() + ",offset=" + recordMetadata.offset() + ",timestamp=" + recordMetadata.timestamp());
            }
        });
        appResult.put("status","SUCCESS");
        return appResult;
    }

    /**
     * 消息有key,按key分区发送
     * @param topic
     * @param value
     * @return
     */
    @RequestMapping("/sendWithKey")
    @ResponseBody
    public Map<String,Object> sendWithKey(String topic,String key,String value,Integer count) {

        if(count == null){
            count = 10;
        }
        Map<String,Object> appResult = new HashMap<>();
        String realKey = key;
        for(int i = 0;i < count;i ++){
            if(StringUtils.isEmpty(key)){
                realKey = i + "";
            }
            ProducerRecord<String,String> producerRecord = new ProducerRecord(topic,realKey,i + "-" + value);
            kafkaProducer.send(producerRecord, new TestSendWithKeyCallback(realKey,i + "-" + value));
        }
        System.out.println("======================================发完啦==============================================");
        appResult.put("status","SUCCESS");
        return appResult;
    }

    class TestSendWithKeyCallback implements Callback{
        String key;
        String value;

        public TestSendWithKeyCallback(String key, String value) {
            this.key = key;
            this.value = value;
        }

        @Override
        public void onCompletion(RecordMetadata recordMetadata, Exception e) {
            System.out.println("发送消息完毕,RecordMetadata:topic=" + recordMetadata.topic() + ",partition=" + recordMetadata.partition() + ",offset=" + recordMetadata.offset() + ",timestamp=" + recordMetadata.timestamp() + ",key=" + key + ",value=" + value);
        }
    }

    /**
     * 发送墓碑消息
     * @param topic
     * @param value
     * @return
     */
    @RequestMapping("/sendTombstone")
    @ResponseBody
    public Map<String,Object> sendTombstone(String topic,String key) {

        Map<String,Object> appResult = new HashMap<>();
        ProducerRecord<String,String> producerRecord = new ProducerRecord(topic,key,null);
        kafkaProducer.send(producerRecord, new Callback() {
            @Override
            public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                System.out.println("发送墓碑消息完毕,RecordMetadata:topic=" + recordMetadata.topic() + ",partition=" + recordMetadata.partition() + ",offset=" + recordMetadata.offset() + ",timestamp=" + recordMetadata.timestamp());
            }
        });
        appResult.put("status","SUCCESS");
        return appResult;
    }
}
